package com.edu.yx.app.App_03_DwdApp.Dwd_01_Traffic;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.edu.yx.app.APP_01_BaseApp.BaseApp;
import com.edu.yx.util.TempUtil;
import com.edu.yx.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Collections;
import java.util.List;

import static com.edu.yx.common.Constant.*;

public class Dwd_01_Traffic_UVDetail extends BaseApp {

    public static void main(String[] args) {

         new Dwd_01_Traffic_UVDetail().init(
            3002,
                2,
                "Dwd_01_Traffic_UVDetail",
                TOPIC_DWD_TRAFFIC_PAGE
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {

        SingleOutputStreamOperator<JSONObject> etlstream = etl(stream);


        //需求解读：一个用户访问1次页面，最早访问的时间，算做对该页面的第一次访问（这里只计算首次访问时间，不考虑访问哪个页面）
        etlstream
//                .map(str-> JSON.parseObject(str)) //将kafka来的String转成json
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                        .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((jsonObject,ts)-> jsonObject.getLong("ts"))
                )
                //根据uid进行keyBy，后续一个窗口的对同一个用户进行处理
                .keyBy(jsonObject-> jsonObject.getJSONObject("common").getString("uid"))
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .process(new ProcessWindowFunction<JSONObject, String, String, TimeWindow>() {

                    private ValueState<String> visitDate;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //记录访问日期，如果不是今天的，那么说明已经存在跨天
                        visitDate = getRuntimeContext().getState(new ValueStateDescriptor<String>("VisitDate", String.class));
                    }

                    @Override
                    public void process(String s,
                                        Context context,
                                        Iterable<JSONObject> elements,
                                        Collector<String> out) throws Exception {

                        //获取状态内的日期
                        String date = visitDate.value();
                        //获取进来数据的日期,窗口开启时间
                        String today = TempUtil.toDate(context.window().getStart());

                        //如果今天的日期与状态相同，那么就不用计算了，因为今天的已经计算过了，否则状态不会更新
                        //如果今天的日期与状态不同，说明两种情况：
                        //              第一：app第一次启动，状态为null，那么与today获取到窗口内的真实日期不同
                        //              第二：跨天的情况，today是17号，但是状态没更新，还是16号

                        //这个状态只针对跨天或者首日计算
                        if (!today.equals(date)){

                            //todo 关键获取一个窗口的同一个key的所有数据，封装到一个list中
                            List<JSONObject> list = TempUtil.toList(elements);

                            JSONObject min = Collections.min(list, (o1, o2) -> o1.getLong("ts").compareTo(o2.getLong("ts")));

                            out.collect(min.toJSONString());

                            //这里更新状态，把今天日期放到的状态内，后续数据同一天的数据就进不来
                            visitDate.update(today);
                        }
                    }
                })
//                .print();
                .addSink(FlinkSinkUtil.getKafkaSink(TOPIC_DWD_TRAFFIC_UV_DETAIL));
    }

    private SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> stream) {

        //只要json数据
        return stream
                .filter(jsonString -> {
                    if (jsonString.equals("")) {
                        System.out.println(jsonString + "jsonString是一个空字符串，JSON.parseObject(jsonString)会将其转化成null");
                        return false;
                    }

                    try {
                        JSON.parseObject(jsonString);
                    } catch (Exception e) {
                        System.out.println("json格式有误，请检查：" + jsonString);
                        e.printStackTrace();
                        return false;
                    }
                    return true;
                })
//                .map(json->{return JSON.parseObject(json);})
                .map(JSON::parseObject);
//                .print();


    }

}
